// readers/dameng/dameng_reader.go
package readers

import (
	"context"
	"database/sql"
	"fmt"
	"mig2kwdb/common"
	"strings"

	_ "gitee.com/chunanyong/dm"
)

type DamengReader struct {
	db        *sql.DB
	connInfo  common.DBConnectionInfo
	dataBases []common.DatabaseConfig
	handler   common.ReplicateHandlerInterface
}

// NewDamengReader 创建新的 DamengReader 实例并初始化数据库连接
func NewDamengReader(connInfo common.DBConnectionInfo) (*DamengReader, error) {
	// 达梦数据库连接字符串格式
	dsn := fmt.Sprintf("dm://%s:%s@%s:%d",
		connInfo.User, connInfo.Password, connInfo.Ip, connInfo.Port)

	dsn += "?connectTimeout=10000&compatibleMode=Mysql"

	db, err := sql.Open("dm", dsn)
	if err != nil {
		return nil, err
	}

	// 测试连接
	if err = db.Ping(); err != nil {
		db.Close()
		return nil, err
	}

	return &DamengReader{
		db:       db,
		connInfo: connInfo,
	}, nil
}

// GetDataBases retrieves a list of all databases in the Dameng instance
func (r *DamengReader) GetDataBases() ([]string, error) {
	// 达梦数据库中模式(Schema)相当于其他数据库的数据库概念
	query := "SELECT DISTINCT OWNER FROM ALL_OBJECTS WHERE OBJECT_TYPE = 'TABLE'"

	rows, err := r.db.Query(query)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var databases []string
	for rows.Next() {
		var schemaName string
		if err := rows.Scan(&schemaName); err != nil {
			return nil, err
		}
		databases = append(databases, schemaName)

	}

	if err = rows.Err(); err != nil {
		return nil, err
	}

	return databases, nil
}

// GetDatabaseTables retrieves a list of all tables in the specified schema
func (r *DamengReader) GetDatabaseTables(databaseName, schemaName string) ([]string, error) {
	// 在达梦数据库中，databaseName参数实际上是模式名
	query := `
		SELECT TABLE_NAME 
		FROM ALL_TABLES 
		WHERE OWNER = UPPER(?)
		ORDER BY TABLE_NAME
	`

	rows, err := r.db.Query(query, databaseName)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var tables []string
	for rows.Next() {
		var tableName string
		if err := rows.Scan(&tableName); err != nil {
			return nil, err
		}
		tables = append(tables, tableName)
	}

	if err = rows.Err(); err != nil {
		return nil, err
	}

	return tables, nil
}

// GetTableColumn retrieves column information for a specific table in a Dameng database
func (r *DamengReader) GetTableColumn(databaseName, schemaName, tableName string) ([]common.KwdbColumn, error) {
	// 达梦数据库获取表列信息的查询
	query := `
		SELECT 
			C.COLUMN_NAME,
			C.COLUMN_ID,
			C.DATA_TYPE,
			COALESCE(C.DATA_LENGTH, 0),
			COALESCE(C.DATA_PRECISION, 0),
			COALESCE(C.DATA_SCALE, 0),
			CASE WHEN C.NULLABLE = 'Y' THEN 'YES' ELSE 'NO' END,
			COALESCE(C.DATA_DEFAULT, ''),
			CASE WHEN C.COLUMN_NAME IN (
				SELECT ACC.COLUMN_NAME 
				FROM ALL_CONS_COLUMNS ACC
				JOIN ALL_CONSTRAINTS AC ON ACC.CONSTRAINT_NAME = AC.CONSTRAINT_NAME
				WHERE ACC.OWNER = C.OWNER AND ACC.TABLE_NAME = C.TABLE_NAME AND AC.CONSTRAINT_TYPE = 'P'
			) THEN 'YES' ELSE 'NO' END,
			CASE WHEN C.COLUMN_NAME IN (
				SELECT ACC.COLUMN_NAME 
				FROM ALL_CONS_COLUMNS ACC
				JOIN ALL_CONSTRAINTS AC ON ACC.CONSTRAINT_NAME = AC.CONSTRAINT_NAME
				WHERE ACC.OWNER = C.OWNER AND ACC.TABLE_NAME = C.TABLE_NAME AND AC.CONSTRAINT_TYPE = 'U'
			) THEN 'YES' ELSE 'NO' END,
			COALESCE(CC.COMMENTS, ''),
			CASE WHEN EXISTS (
				SELECT 1
				FROM ALL_TAB_COLUMNS ATC 
				WHERE ATC.OWNER = C.OWNER 
				AND ATC.TABLE_NAME = C.TABLE_NAME 
				AND ATC.COLUMN_NAME = C.COLUMN_NAME 
				AND ATC.DATA_TYPE = 'IDENTITY'
			) THEN 'YES' ELSE 'NO' END,
			'',
			''
		FROM ALL_TAB_COLUMNS C
		LEFT JOIN ALL_COL_COMMENTS CC ON C.OWNER = CC.OWNER AND C.TABLE_NAME = CC.TABLE_NAME AND C.COLUMN_NAME = CC.COLUMN_NAME
		WHERE C.OWNER = UPPER(?) AND C.TABLE_NAME = UPPER(?)
		ORDER BY C.COLUMN_ID
	`

	rows, err := r.db.Query(query, databaseName, tableName)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var columns []common.KwdbColumn
	for rows.Next() {
		var col common.KwdbColumn

		err := rows.Scan(
			&col.Name,
			&col.Position,
			&col.Type,
			&col.Length,
			&col.Precision,
			&col.Scale,
			&col.Nullable,
			&col.DefaultValue,
			&col.PrimaryKey,
			&col.Unique,
			&col.Comment,
			&col.AutoIncrement,
			&col.Charset,
			&col.Collation,
		)
		if err != nil {
			return nil, err
		}

		columns = append(columns, col)
	}

	if err = rows.Err(); err != nil {
		return nil, err
	}

	return r.MapDataTypes(columns), nil
}

// MapDataTypes 映射达梦数据类型到通用类型
func (r *DamengReader) MapDataTypes(column []common.KwdbColumn) []common.KwdbColumn {
	for i := range column {
		switch strings.ToUpper(column[i].Type) {
		case "REAL":
			column[i].Type = common.DataTypes.FLOAT8.Name
		case "BIT":
			column[i].Type = common.DataTypes.BOOL.Name
		case "VARCHAR2":
			column[i].Type = common.DataTypes.VARCHAR.Name
		case "NUMBER":
			column[i].Type = common.DataTypes.DECIMAL.Name
		case "VARBINARY", "BINARY", "IMAGE":
			column[i].Type = common.DataTypes.BYTES.Name
		case "LONGVARCHAR":
			column[i].Type = common.DataTypes.TEXT.Name
		case "LONGVARBINARY":
			column[i].Type = common.DataTypes.BLOB.Name
		case "BYTE":
			column[i].Type = common.DataTypes.BYTES.Name
		case "DATETIME":
			column[i].Type = common.DataTypes.TIMESTAMP.Name
		case "DATETIME WITH TIME ZONE", "TIMESTAMP WITH LOCAL TIME ZONE":
			column[i].Type = common.DataTypes.TIMESTAMPTZ.Name
		}
	}
	return column

}

// GetTableDefinition retrieves the definition of a table including its columns
func (r *DamengReader) GetTableDefinition(databaseName, schemaName, tableName string) (common.TableDefinition, error) {
	var def common.TableDefinition
	def.DatabaseName = databaseName
	def.SchemaName = schemaName
	def.TableName = tableName

	var err error
	def.Columns, err = r.GetTableColumn(databaseName, schemaName, tableName)
	if err != nil {
		return def, err
	}

	return def, nil
}

// GetConditionRows retrieves rows from a table with optional condition
func (r *DamengReader) GetConditionRows(databaseName, schemaName, tableName string, columns []common.KwdbColumn, condition string) (common.RowsInterface, error) {
	orderBy := ""
	keyNum := 0
	sqlQuery := "SELECT "

	for idx, col := range columns {
		if idx > 0 {
			sqlQuery += ","
		}
		sqlQuery += " \"" + col.Name + "\""

		// 确定排序字段（主键优先）
		if col.PrimaryKey.String == "YES" {
			if keyNum > 0 {
				orderBy += ","
			}
			orderBy += " \"" + col.Name + "\""
			keyNum++
		}
	}

	// 如果没有主键，则使用唯一索引
	if orderBy == "" {
		keyNum = 0
		for _, col := range columns {
			if col.Unique.String == "YES" {
				if keyNum > 0 {
					orderBy += ","
				}
				orderBy += " \"" + col.Name + "\""
				keyNum++
			}
		}
	}

	sqlQuery += " FROM \"" + databaseName + "\".\"" + tableName + "\""

	if condition != "" {
		sqlQuery += " WHERE " + condition
	}

	if orderBy != "" {
		sqlQuery += " ORDER BY " + orderBy
	}

	rows, err := r.db.Query(sqlQuery)
	return common.WrapSQLRows(rows), err
}

// GetDB returns the underlying database connection
func (r *DamengReader) GetDB() *sql.DB {
	return r.db
}

// Close closes the database connection
func (r *DamengReader) Close() {
	if r.db != nil {
		r.db.Close()
	}
}

// GetDataBaseType returns the type of the database
func (r *DamengReader) GetDataBaseType() string {
	return "dameng"
}

// IsNetworkError checks if an error is a network error
func (r *DamengReader) IsNetworkError(err error) bool {

	return true
}

// GetStartPos 获取复制起始位置（达梦数据库需要启用归档日志和补充日志）
func (r *DamengReader) GetStartPos() (common.ReplicateStartPos, error) {
	var startPos common.ReplicateStartPos

	return startPos, nil
}

// StartReplicate 开始复制过程
func (r *DamengReader) StartReplicate(ctx context.Context, startPos common.ReplicateStartPos) error {

	return nil
}

// SetDataBases 设置需要同步的数据库配置
func (r *DamengReader) SetDataBases(dataBases []common.DatabaseConfig) {
	r.dataBases = dataBases
}

// SetReplicateHandler 设置复制处理器
func (r *DamengReader) SetReplicateHandler(handler common.ReplicateHandlerInterface) {
	r.handler = handler
}
